package com.xiguaji.tiktok.task.market.service;


import com.gugee.tiktok.common.model.spider.market.author.Author;
import com.gugee.tiktok.common.model.spider.market.author.MarketAuthorResponse;
import com.gugee.tiktok.data.common.msg.MarketMessage;
import com.xiguaji.tiktok.task.common.config.AppConfig;
import com.xiguaji.tiktok.task.market.spider.MarketSpiderClient;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.client.RestTemplate;

import javax.annotation.PostConstruct;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: luolifeng
 * Date: 2020-11-16 9:57
 */
 @Getter
 @Slf4j
public class MarkerForkJoinTask extends RecursiveTask<Integer> {
    //每一个子任务中处理的记录数
    int maxRecords=2000;
    MarketSpiderClient marketSpiderClient;
    RestTemplate restTemplate;
    int limit=20;
    @Value("${xiguaji.spider.host}")
    String host;

    AppConfig appConfig;

    private   KafkaTemplate<Long, MarketMessage> kafkaTemplateMarket;

    MarketForkJoinParam params;
    private String topic;
    private final static AtomicInteger errorCnt=new AtomicInteger(0);
    public void init(){
        errorCnt.set(0);
    }

    public MarkerForkJoinTask(MarketForkJoinParam params,MarketSpiderClient marketSpiderClient,RestTemplate  restTemplate,KafkaTemplate<Long, MarketMessage> kafkaTemplateMarket, AppConfig appConfig) {
        this.maxRecords=params.getMaxRecords();
        this.marketSpiderClient =marketSpiderClient;
        this.params =params;
        this.topic =params.getTopic();
        this.restTemplate =restTemplate;
        this.appConfig =appConfig;
        this.kafkaTemplateMarket =kafkaTemplateMarket;
    }

    @Override
    protected Integer compute() {

        MarketAuthorResponse marketAuthorResponse=marketSpiderClient.getAuthorSearchByFans(1,limit,params.getStartValue(),params.getEndValue());
        int cnt =marketAuthorResponse.getData().getPagination().getTotal_count();


        if(cnt ==0 ) {
            return 0;
        }
        // 如果条件成立，说明这个任务可以开始执行了
        if(cnt <= maxRecords ) {
            String inf="开始计算的部分：startValue = " + params.getStartValue() + ";endValue = " + params.getEndValue()+";totalcnt="+cnt;
            XxlJobLogger.log(inf);
            long start=System.currentTimeMillis();
            int tmpErr=0;

            int totalPages=marketAuthorResponse.getData().getPagination().getTotal_count()/limit+1;
            for(int i=1;i<=totalPages;i++) {
                MarketAuthorResponse response=marketSpiderClient.getAuthorSearchByFans(i,limit,params.getStartValue(),params.getEndValue());
                for (Author author : response.getData().getAuthors()) {
                    try {
//                        log.info("appConfig={},params={},kafkaTemplateMarket={}", appConfig,params, kafkaTemplateMarket);
                        MarketMessage marketMessage=new MarketMessage();
                        marketMessage.setAuthor(author);
                        marketMessage.setRetryLimitTimes(appConfig.getRetryMarketTimes());
                        marketMessage.setRetryTimes(appConfig.getRetryMarketTimes());
                        kafkaTemplateMarket.send(params.getTopic(),marketMessage);
                    } catch (Exception e) {
                        errorCnt.incrementAndGet();
                        tmpErr++;
                        log.error("error={},uid={}", e.getMessage(), author.getCore_user_id());
                        XxlJobLogger.log("error={},uid={}", e.getMessage(), author.getCore_user_id());
                    }
                }

            }
            XxlJobLogger.log("更新成功:startValue={},endValue={},successCnt={},errorCnt={},耗时={}", params.getStartValue(),params.getEndValue(),cnt,tmpErr,(System.currentTimeMillis() - start) / 1000);
            return cnt;
        }
        // 否则再进行任务拆分，拆分成两个任务
        MarketForkJoinParam subParams1=new MarketForkJoinParam(params.getStartValue(),(params.getStartValue() + params.getEndValue()) / 2,params.getMaxRecords(),params.getTopic());
        MarkerForkJoinTask subTask1 = new MarkerForkJoinTask(subParams1,marketSpiderClient,restTemplate,kafkaTemplateMarket,appConfig);
            subTask1.fork();
        MarketForkJoinParam subParams2=new MarketForkJoinParam((params.getStartValue() + params.getEndValue()) / 2 + 1,params.getEndValue(),params.getMaxRecords(),params.getTopic());
        MarkerForkJoinTask subTask2 = new MarkerForkJoinTask(subParams2 , marketSpiderClient,restTemplate,kafkaTemplateMarket,appConfig);
            subTask2.fork();
        return subTask1.join() + subTask2.join();

    }


    @Data
    @AllArgsConstructor
    public static class MarketForkJoinParam{
        // 子任务开始计算的粉丝范围值
        Integer startValue;
        // 子任务结束计算的粉丝范围值
        Integer endValue;
        int maxRecords;
        String topic;

    }
}

